package edu.hubu.wjw

import org.apache.flink.api.common.serialization.SimpleStringEncoder
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.Path
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util.UUID
import java.util.concurrent.TimeUnit
import scala.util.Random

object OrderCountDemo {

  // 样例类(订单编号，手机编号，手机名称，手机价格，订单时间)
  case class Order(orderID: String, itemID: String, phoneName: String, price: Int, time: Long)

  def main(args: Array[String]): Unit = {
    // TODO:1.Environment
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // TODO:2.Source
    val sourceDStream: DataStream[Order] = env.addSource(new OrderSourceFunc)
    sourceDStream.print()

    // TODO:4.Transformation
    // 实时统计各手机销售总金额
    val resultDStream: DataStream[(String, String, Int, Long)] = sourceDStream
      .keyBy(_.itemID)
      .reduce((preData, curData) =>
        Order(preData.orderID, preData.itemID, preData.phoneName, preData.price + curData.price, curData.time))
      .map(data => (data.itemID, data.phoneName, data.price, data.time))

    // 结果写入MySQL
    resultDStream.addSink(new MySQLSinkFunc)


    // TODO:4.Sink
    // 将生成的手机订单信息写入本地文件系统
    //    sourceDStream.writeAsText("src/main/resources/phoneOrder2.txt")
    sourceDStream.addSink(StreamingFileSink
      .forRowFormat(new Path("src/main/resources/phoneOrder2.txt"), new SimpleStringEncoder[Order]())
      .build()
    )

    // TODO:5.Execution
    env.execute("OrderCountDemo Job")

  }

  /**
   * @describe 生成手机订单数据
   */
  class OrderSourceFunc extends SourceFunction[Order] {
    //定义变量
    private var count: Long = 0L
    private var isRunning: Boolean = true
    private val nameList: List[String] = List("Redmi K40", "iPhone 13 Pro", "小米11 Pro", "Redmi Note 10 Pro", "Huawei P50", "荣耀 60")
    private val priceList: List[Int] = List(2399, 8699, 4398, 1399, 4299, 2689)
    private val random: Random = new Random()

    override def run(ctx: SourceFunction.SourceContext[Order]): Unit = {
      while (isRunning && count < 1000) {
        // 订单编号，手机编号，手机名称，手机价格，订单时间
        val orderID: String = UUID.randomUUID().toString
        val i: Int = Random.nextInt(6)
        val itemID: String = "item" + i
        val phoneName: String = nameList(i)
        val price: Int = priceList(i)
        val time: Long = System.currentTimeMillis()

        // 收集生成的数据
        ctx.collect(Order(orderID, itemID, phoneName, price, time))

        // 统计生成的订单数
        count += 1

        // 每隔1秒生成一个订单
        TimeUnit.SECONDS.sleep(1)
        // 每隔100毫秒生成一个订单
        //        TimeUnit.MILLISECONDS.sleep(100)
      }
    }

    override def cancel(): Unit = {
      isRunning = false
    }
  }

  /**
   * 将处理好的结果数据写入MySQL数据库
   */
  class MySQLSinkFunc extends RichSinkFunction[(String, String, Int, Long)] {
    var connection: Connection = null
    var ps: PreparedStatement = null

    override def open(parameters: Configuration): Unit = {
      //加载驱动，打开连接
      Class.forName("com.mysql.jdbc.Driver")
      connection = DriverManager.getConnection(
        "jdbc:mysql://localhost:3306/******?useSSL=false&characterEncoding=utf8",
        "root",
        "******")

      ps = connection.prepareStatement("update salesamount set amount=?, time=? where itemID=?")
    }

    override def invoke(value: (String, String, Int, Long), context: SinkFunction.Context): Unit = {
      //元组的主键，取系统此时的时间戳
      ps.setInt(1, value._3)
      ps.setLong(2, value._4)
      ps.setString(3, value._1)
      //执行sql语句
      ps.executeUpdate()

    }

    override def close(): Unit = {
      if (connection != null)
        connection.close()
      if (ps != null)
        ps.close()
    }

  }

}
